package com.devilvan.betrice.config;

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import reactor.core.publisher.Flux;
import reactor.util.function.Tuple2;

import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;
import java.util.function.Function;

/**
 * @author Evad.Wu
 * @Description mq消息通道 配置类
 * @date 2023-02-11
 */
@Configuration
public class BetriceMqSubChannel {
//    /**
//     * Supplier声明一个消息生产者，source就对应source-out-0的生产者binding
//     */
//    @Bean
//    public Supplier<String> source() {
//        return () -> {
//            String message = "FromSource1";
//            System.out.println("******************");
//            System.out.println("From Source1");
//            System.out.println("******************");
//            System.out.println("Sending value: " + message);
//            return message;
//        };
//    }

    /**
     * Function方式声明binding，相当于同时声明了一个Producer的Bindng和一个Consumer的Binding。
     * transfer-in-0 表示消费者  transfer-out-0 表示生产者
     * 其作用就在于，echo-in-0收到的消息，立即就会通过echo-out-0发送出去。
     */
    @Bean
    public Function<String, String> transfer() {
        return message -> {
            try {
                TimeUnit.SECONDS.sleep(2);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("transfer: " + message);
            return "transfer:" + message;
        };
    }

    /**
     * Consumer声明一个消息消费者，sink1就对应sink-in-0
     */
    @Bean
    public Consumer<String> sink() {
        return message -> {
            System.out.println("******************");
            System.out.println("At Sink1");
            System.out.println("******************");
            System.out.println("Received message " + message);
        };
    }

    /**
     * 例如这样的Function，就需要声明gather-in-0和gather-in-1。
     * 如果输出的是个Tuple，那就同样需要声明gather-out-0和gather-out-1
     */
    @Bean
    public Function<Tuple2<Flux<String>, Flux<String>>, Flux<String>> gather() {
        return tuple -> {
            Flux<String> t1 = tuple.getT1();
            Flux<String> t2 = tuple.getT2().map(str -> "gather2:" + str);
//            return Flux.merge(t1,t2);

            return Flux.combineLatest(t1, t2, (str1, str2) -> "gather1:" + str1 + ";" + str2);
        };
    }

    @Bean
    public Consumer<String> gatherEcho() {
        return message -> {
            System.out.println("received message from gatherCosuer : " + message);
        };
    }
}
